package com.atguigu.app.dwd;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.utils.MyKafkaUtil;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

//数据流： web/app -> Nginx -> 日志服务器(xx.log) -> Flume -> Kafka(ODS) -> FlinkApp -> Kafka(DWD)
//程  序： Mock -> Flume(f1.sh) -> Kafka(ZK) -> BaseLogApp -> Kafka(ZK)
public class BaseLogApp {

    public static void main(String[] args) throws Exception {

        //TODO 1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);  //生产环境设置为Kafka分区数

        //        //1.1 开启CheckPoint
        //        env.enableCheckpointing(5 * 60000L);
        //        env.getCheckpointConfig().setCheckpointTimeout(5 * 60000L);
        //        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(10000L);
        //        env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
        //        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3,5000L));
        //
        //        //1.2 指定状态后端
        //        env.setStateBackend(new FsStateBackend("hdfs://hadoop102:8020/xxxx/xxx"));

        //TODO 2.读取Kafka topic_log 主题的数据创建流
        DataStreamSource<String> kafkaDS = env.addSource(MyKafkaUtil.getKafkaConsumer("topic_log", "base_log_app_210927"));

        //TODO 3.将数据转换为JSON对象,非JSON数据写入侧输出流
        OutputTag<String> dirtyTag = new OutputTag<String>("DirtyData") {
        };
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaDS.process(new ProcessFunction<String, JSONObject>() {
            @Override
            public void processElement(String value, Context ctx, Collector<JSONObject> out) throws Exception {
                try {
                    JSONObject jsonObject = JSON.parseObject(value);
                    out.collect(jsonObject);
                } catch (Exception e) {
                    ctx.output(dirtyTag, value);
                }
            }
        });
        //获取脏数据并打印
        jsonObjDS.getSideOutput(dirtyTag)
                .print("Dirty>>>>>>>");

        //TODO 4.新老用户校验(根据Mid分组,状态编程)
        KeyedStream<JSONObject, String> keyedStream = jsonObjDS.keyBy(jsonObj -> jsonObj.getJSONObject("common").getString("mid"));
        SingleOutputStreamOperator<JSONObject> jsonObjWithNewFlagDS = keyedStream.map(new RichMapFunction<JSONObject, JSONObject>() {

            private ValueState<String> valueState;

            @Override
            public void open(Configuration parameters) throws Exception {
                valueState = getRuntimeContext().getState(new ValueStateDescriptor<String>("first", String.class));
            }

            @Override
            public JSONObject map(JSONObject value) throws Exception {

                //获取is_new标记
                String isNew = value.getJSONObject("common").getString("is_new");

                //判断标记是否为"1"
                if ("1".equals(isNew)) {

                    //取出状态数据
                    String firstVisit = valueState.value();

                    //判断状态是否为Null
                    if (firstVisit == null) {
                        //更新状态
                        valueState.update("0");
                    } else {
                        //将数据中标记改为"0"
                        value.getJSONObject("common").put("is_new", "0");
                    }
                }

                return value;
            }
        });

        //打印测试
        //jsonObjWithNewFlagDS.print("jsonObjWithNewFlagDS>>>>>>>>>>");

        //TODO 5.分5个流(侧输出流) 页面日志写入主流  其他(启动，曝光，动作，错误)则写入侧输出流
        OutputTag<String> startTag = new OutputTag<String>("start") {
        };
        OutputTag<String> displayTag = new OutputTag<String>("display") {
        };
        OutputTag<String> actionTag = new OutputTag<String>("action") {
        };
        OutputTag<String> errorTag = new OutputTag<String>("error") {
        };
        SingleOutputStreamOperator<String> pageDS = jsonObjWithNewFlagDS.process(new ProcessFunction<JSONObject, String>() {
            @Override
            public void processElement(JSONObject value, Context ctx, Collector<String> out) throws Exception {

                //将数据转换为字符串
                String valueString = value.toJSONString();

                //尝试获取错误字段
                String err = value.getString("err");
                if (err != null) {
                    ctx.output(errorTag, valueString);
                }

                //尝试获取启动字段
                String start = value.getString("start");
                if (start != null) {
                    ctx.output(startTag, valueString);
                } else {
                    out.collect(valueString);

                    //提取数据中的时间戳以及page_id
                    Long ts = value.getLong("ts");
                    String pageId = value.getJSONObject("page").getString("page_id");

                    //尝试获取曝光数据
                    JSONArray displays = value.getJSONArray("displays");
                    if (displays != null && displays.size() > 0) {
                        //遍历单个曝光数据并写出
                        for (int i = 0; i < displays.size(); i++) {
                            JSONObject display = displays.getJSONObject(i);
                            display.put("page_id", pageId);
                            display.put("ts", ts);

                            ctx.output(displayTag, display.toJSONString());
                        }
                    }

                    //尝试获取曝光数据
                    JSONArray actions = value.getJSONArray("actions");
                    if (actions != null && actions.size() > 0) {
                        //遍历单个曝光数据并写出
                        for (int i = 0; i < actions.size(); i++) {
                            JSONObject action = actions.getJSONObject(i);
                            action.put("page_id", pageId);
                            action.put("common", value.getString("common"));

                            ctx.output(actionTag, action.toJSONString());
                        }
                    }
                }
            }
        });

        //TODO 6.获取每个流的数据并写入Kafka DWD层
        DataStream<String> startDS = pageDS.getSideOutput(startTag);
        DataStream<String> actionDS = pageDS.getSideOutput(actionTag);
        DataStream<String> displayDS = pageDS.getSideOutput(displayTag);
        DataStream<String> errorDS = pageDS.getSideOutput(errorTag);

        pageDS.print("Page>>>>>>>>");
        startDS.print("Start>>>>>>>>");
        actionDS.print("Action>>>>>>>>");
        displayDS.print("Display>>>>>>>>");
        errorDS.print("Error>>>>>>>>");

        pageDS.addSink(MyKafkaUtil.getKafkaProducer("dwd_page_log"));
        startDS.addSink(MyKafkaUtil.getKafkaProducer("dwd_start_log"));
        actionDS.addSink(MyKafkaUtil.getKafkaProducer("dwd_action_log"));
        displayDS.addSink(MyKafkaUtil.getKafkaProducer("dwd_display_log"));
        errorDS.addSink(MyKafkaUtil.getKafkaProducer("dwd_error_log"));

        //TODO 7.启动任务
        env.execute("BaseLogApp");

    }

}
